package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

/**
 * @author Felix
 * @date 2023/12/18
 * 将维度数据写到Hbase中
 */
public class DimSinkFunction  extends RichSinkFunction<JSONObject> {
    private Connection hbaseConn;
    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(hbaseConn);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //{"tm_name":"Redmi111","id":1,"type":"update",
        //"table_process":{"op":"r","sinkColumns":"id,tm_name","sinkFamily":"info","sinkRowKey":"id",
        //"sinkTable":"dim_base_trademark","sourceTable":"base_trademark"}}

        //获取业务数据库中维度表的操作类型
        String type = jsonObj.getString("type");
        //获取操作的hbase表的表名
        TableProcess tableProcess = jsonObj.getObject("table_process", TableProcess.class);
        String sinkTable = tableProcess.getSinkTable();

        //获取要从hbase表中删除的数据对应的rowkey
        String rowKeyName = tableProcess.getSinkRowKey();
        String rowkeyValue = jsonObj.getString(rowKeyName);

        if("delete".equals(type)){
            //说明对业务数据库的维度表进行了删除操作，对hbase对应的维度表也应该执行删除
            System.out.println("从Hbase的"+sinkTable+"表中删除了"+rowkeyValue+"数据");
            HbaseUtil.deleteRow(hbaseConn, GmallConfig.HBASE_NAMESPACE,sinkTable,rowkeyValue);
        }else{
            //说明对业务数据库的维度表进行了删除外的其它操作(insert|update|bootstart-insert),将最新的维度数据put到hbase表中
            String sinkFamily = tableProcess.getSinkFamily();
            String[] columnNames = tableProcess.getSinkColumns().split(",");
            //创建一个和列名对应的列值数组
            String[] columnValues = new String[columnNames.length];
            for (int i = 0; i < columnNames.length; i++) {
                String columnName = columnNames[i];
                columnValues[i] = jsonObj.getString(columnName);
            }
            System.out.println("向Hbase的"+sinkTable+"表中添加了"+rowkeyValue+"数据");

            HbaseUtil.putRow(hbaseConn,GmallConfig.HBASE_NAMESPACE,sinkTable,rowkeyValue,sinkFamily,
                columnNames,columnValues);
        }

        //将Redis中缓存的维度数据删除掉
        if("delete".equals(type)||"update".equals(type)){
            DimUtil.delCached(sinkTable,rowkeyValue);
        }
    }
}
